package com.study.spark.scala.sql

import org.apache.spark.sql.SparkSession

/**
 * 跨列处理的函数
 *
 * @author stephen
 * @date 2019-09-27 18:43
 */
object MultipleColumnsFunctionDemo {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("FilterWhereWhenDemo")
      .master("local[*]")
      .getOrCreate()

    // 隐式转换
    import org.apache.spark.sql.functions._
    import spark.implicits._

    val df1 = spark.createDataset(Seq((null, "2"), ("0", null), (null, null), ("4", "5"))).toDF("key1", "key2")
    // 造点NaN的字段
    val df = df1.withColumn("key3", sqrt(rand() - 0.4))
    // 要找出给定的几个列中的第一个非空值
    df.select($"*", coalesce($"key1", $"key2", $"key3")).show()
    // nanvl函数来处理NaN值
    df.select($"key2", $"key3", nanvl($"key2", $"key3")).show()
    // greatest()函数，找出给定所有列中的最大值，同样的，NaN有些特别
    df.select($"*", greatest($"key1".cast("double"), $"key2".cast("double"), $"key3")).show()
    // 找最小值时，NaN就不见了，可知NaN是被当成一个极大值处理的
    df.select($"*", least($"key1".cast("double"), $"key2".cast("double"), $"key3")).show()

    spark.stop()
  }

}
